Python で DI(Dependency Injection) を実現するフレームワークの Injector を使ってみる

Python コードの品質向上のために、DI(Dependency Injection) フレームワークの injector を導入してみました。

#Python

#AWS Lambda

Takaaki Kakei

2022.11.21

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

こんにちは、筧( @TakaakiKakei )です。

所属しているチームでは、開発言語として python をよく使っています。そして最近、コード品質向上のために injector を導入しました。

alecthomas/injector: Python dependency injection framework, inspired by Guice

injector は python で DI を実現するフレームワークです。私は DI 初心者で、同僚のコードを読み解きながら理解を進めている状況です。そこで今回は学んだことを、本ブログにアウトプットします。

何が嬉しいのか

嬉しいことの一つとしてテストが書きやすくなることが挙げられます。 injector を導入することで、外部からのパラメータの注入や、外部操作するインスタンス生成を Module クラスなどに集めることができます。これによって、「 Module クラスのパラメータをMockすれば単体で動かせるぞ」という状況を作りやすくなり、テストが書きやすくなることが期待できます。

前提

所属チームで採用している階層構造について

従来

injector 採用前の階層構造です。

.
└── src
    ├── __init__.py
    ├── handlers
    │   ├── __init__.py
    │   ├── hello.py
    ├── services
    │   ├── __init__.py
    │   └── slack.py
    └── use_cases
        ├── __init__.py
        └── hello.py

各階層の役割

以下が各階層の役割です。

handlers
- プログラムのエントリーポイントを定義
modules
- 外部からのパラメータの注入や、外部操作するインスタンス生成を定義
services
- slack などのサービスを定義
use_cases
- ビジネスロジックを定義

なお、従来の階層構造では、modules 層の役割を、use_cases 層が担っていました。

DI(Dependency Injection) とは

解釈の一つとして、「オブジェクト注入」と言えます。依存元クラスが具象クラスに依存しているのを、抽象クラスに依存させることで、コンポーネント間の依存関係を削除します。これによって、コードの変更をしやすくなったり、前述のようにテストが書きやすくなるソフトウェアパターンです。以下の資料が読み物としてわかりやすかったので、リンクを置いておきます。

Python で Dependency Injection(DI) をやるには？ - Speaker Deck

injector で利用するアノテーションについて

コード紹介で出てくる、injector のアノテーションについて説明します。

@inject
- オブジェクト注入する際に宣言します。
- 所属チームでは、modles/ services/ use_cases の層の __init__ に付与することが多いです。
- 型宣言が必須です。
@singleton
- インジェクトの度に新しいインスタンスを生成したくない時に宣言します。
- 所属チームでは、modles 層のメソッドと services/ use_cases 層のクラスに付与することが多いです。
@provider
- メソッドと戻り値をインジェクトに利用したい場合に宣言します。
- 所属チームでは、modles 層のメソッドに付与することが多いです。

コード例

前提

event 内の title と text の内容を Slack の Webhook 経由で Slack 通知するだけのサービスです。コード内のアノテーションの意味は前述を参照ください。また以下のライブラリは poetry などでインストール前提です。

python
requests
boto3
injector
aws-lambda-powertools

handlers 層

from typing import Dict

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.utilities.typing import LambdaContext
from injector import Injector

from src.modules.notify_message import NotifyMessageModule
from src.use_cases.notify_message_to_slack import NotifyMessageToSlackUseCase

logger = Logger()
tracer = Tracer()

# Injector()内で、利用する Module を宣言
# Module 内では、外部からのパラメータの注入や、外部操作するインスタンス生成を定義
di = Injector([NotifyMessageModule()])

@tracer.capture_lambda_handler
@logger.inject_lambda_context(log_event=True)
def handler(event: Dict, context: LambdaContext) -> Dict:
    """
    notify_message のエントリーポイント

    event:
    {
        "title": "TEST",
        "text": "hello"
    }
    """
    try:
        # di.get()内で、利用する UseCase を宣言し、各インスタンスを生成
        notify_message_use_case = di.get(NotifyMessageToSlackUseCase)
        # UseCaseのインスタンスメソッドの exec を実行
        notify_message_use_case.exec(event)
        return event
    except Exception as e:
        logger.exception("throw exception in notify_message")
        raise e

プログラムのエントリーポイントを定義しています。ハイライトの箇所が DI に関する主な処理です。
上記コード内のアノテーションは AWS Lambda Powertools 関連なので無視いただいても結構です。
handler 外にdi = Injector(...)を書くことで、singleton ルールで書いてあるものが global 領域のメモリ保持に保持されます（ウォームスタート）。キャッシュされるので、SecretManager の値を更新したのにすぐに反映されないといった事象があるので注意しましょう。もし handler 内に書いた場合は、キャッシュされず、singleton ルールで書いてあるものが lambda に呼ばれる度に作られます。

modules 層

import os

import requests
from injector import Module, provider, singleton

from src.services.slack import SlackSession, SlackWebhookUrl


class NotifyMessageModule(Module):
    @singleton
    @provider
    # src.services.slack の SlackService クラスの __init__ で必要な引数を返す
    def slack_webhook_url(self) -> SlackWebhookUrl:
        return os.environ["SLACK_WEBHOOK_URL"]

    @singleton
    @provider
    # src.services.slack の SlackService クラスの __init__ で必要な引数を返す
    def slack_session(self) -> SlackSession:
        return requests.Session()

外部からのパラメータの注入や、外部操作するインスタンス生成を定義しています。
利用する Service クラスの __init__ で必要な引数を主に返します。

services 層

import json
from typing import Dict, NewType

import requests
from aws_lambda_powertools import Logger
from injector import inject, singleton

# 型エイリアス
SlackWebhookUrl = NewType("SlackWebhookUrl", str)
SlackSession = NewType("SlackSession", requests.Session)

logger = Logger(child=True)


@singleton
class SlackService:
    @inject
    def __init__(self, session: SlackSession, webhook_url: SlackWebhookUrl):
        self._session = session
        self._webhook_url = webhook_url

    def send_msg(self, payload: Dict) -> None:
        data = json.dumps(payload)
        resp = self._session.post(self._webhook_url, data=data)
        logger.info(f"resp.text: {resp.text}")
        logger.info(f"payload: {data}")

外部サービスなどを定義する層で、slack のサービスを定義しています。
ハイライトの箇所を参照すると、NotifyMessageModule で定義していたものがあることが分かります。

use_cases 層

from typing import Dict

from aws_lambda_powertools import Logger
from injector import inject, singleton

from src.services.slack import SlackService

logger = Logger(child=True)


@singleton
class NotifyMessageToSlackUseCase:
    @inject
    def __init__(self, slack_service: SlackService):
        self.slack_service = slack_service

    def exec(self, event: Dict) -> None:
        blocks = [
            {
                "type": "header",
                "text": {"type": "plain_text", "text": event["title"]},
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": event["text"]},
            },
        ]
        payload = {"blocks": blocks}
        self.slack_service.send_msg(payload)
        return

ビジネスロジックを定義しています。payload で送信内容を定義して、インスタンス作成済みの SlackService の send_msg メソッドを使って、Slack チャンネルにメッセージ送信しています。
NotifyMessageToSlackUseCase の初期値は、src/handlers/notify_message.py の31行目の di.get() 時に注入しています。
NotifyMessageToSlackUseCase の exec は、src/handlers/notify_message.py の33行目で実行されます。